package main;

import com.dtstack.flinkx.config.DataTransferConfig;
import com.dtstack.flinkx.reader.DataReader;
import com.dtstack.flinkx.reader.DataReaderFactory;
import com.dtstack.flinkx.writer.DataWriter;
import com.dtstack.flinkx.writer.DataWriterFactory;
import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Preconditions;


public class Main {
    public static void main(String[] args) throws Exception {

        // 解析命令行参数
        Options options = new Options();
        options.addOption("job", true, "Job config.");
//        options.addOption("jobid", true, "Job unique id.");
//        options.addOption("monitor", true, "Monitor Addresses");
        options.addOption("pluginRoot", true, "plugin path root");
        BasicParser parser = new BasicParser();
        CommandLine cl = parser.parse(options, args);
        String job = cl.getOptionValue("job");
        String monitor = cl.getOptionValue("monitor");
        String pluginRoot = cl.getOptionValue("pluginRoot");
        Preconditions.checkNotNull(job, "Must provide --job argument");

        // 解析jobPath指定的任务配置文件
        DataTransferConfig config = DataTransferConfig.parseFromFile(job);

        if(StringUtils.isNotEmpty(monitor)) {
            config.setMonitorUrls(monitor);
        }

        if(StringUtils.isNotEmpty(pluginRoot)) {
            config.setPluginRoot(pluginRoot);
        }


        //构造并执行flink任务
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(config.getJob().getSetting().getSpeed().getChannel());

        DataReader dataReader = DataReaderFactory.getDataReader(config, env);
        DataStream<Row> dataStream = dataReader.readData();
        dataStream = dataStream.rebalance();
        DataWriter dataWriter = DataWriterFactory.getDataWriter(config);
        dataWriter.writeData(dataStream);

//        if(env instanceof MyLocalStreamEnvironment) {
//            List<URL> urlList = new ArrayList<>();
//            URLClassLoader readerClassLoader = (URLClassLoader) dataReader.getClass().getClassLoader();
//            urlList.addAll(Arrays.asList(readerClassLoader.getURLs()));
//            URLClassLoader writerClassLoader = (URLClassLoader) dataWriter.getClass().getClassLoader();
//            for (URL url : writerClassLoader.getURLs()) {
//                if (!urlList.contains(url)) {
//                    urlList.add(url);
//                }
//            }
//            ((MyLocalStreamEnvironment) env).setClasspaths(urlList);
//        }

        env.execute("test flinkx");

    }
}
